package com.shujia.flink.source
import org.apache.flink.streaming.api.functions.source.{ParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._
object Demo2SourceFunction {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * 自定义source
      * 实现sourceFunction接口
      * 实现里面的run方法---> 在run方法中获取数据发送到下游
      */

    val myDS: DataStream[Int] = env.addSource(new MySourceFunction)

    myDS.print()

    env.execute()
  }
}
/**
  *
  * ParallelSourceFunction   : 并行的source, 可以多个线程一个读取数据，需要解决数据重复问题
  * RichParallelSourceFunction: 并行的source，多了open和close 方法
  * RichSourceFunction ：单一的source，多了open和close 方法
  * SourceFunction ： 单一的source
  */

class MySourceFunction extends SourceFunction[Int]{
  /**
    * 在run 方法中获取数据，将数据发送到下游
    *
    *
    * @param ctx flink source的上下文对象，用于将数据发送到下游
    */
  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {
    var i = 0
    while(true){

      //将数据发送到下游
      ctx.collect(i)

      i += 1
      Thread.sleep(1000)
    }
  }

  /**
    * 任务取消的时候调用
    *
    * 用于回收资源
    */
  override def cancel(): Unit = {

  }
}